package communication;

import java.util.ArrayList;
import java.util.List;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.jgroups.Address;
import org.jgroups.Channel;
import org.jgroups.ChannelException;
import org.jgroups.JChannel;
import org.jgroups.MembershipListener;
import org.jgroups.Message;
import org.jgroups.SuspectedException;
import org.jgroups.TimeoutException;
import org.jgroups.View;
import org.jgroups.blocks.MessageDispatcher;
import org.jgroups.blocks.RequestOptions;
import org.jgroups.util.RspList;

import role.Role;

public class Communicator implements org.jgroups.blocks.RequestHandler, MembershipListener, Runnable {
	
	private List<RequestHandler> handlers;

	private Channel channel;
	
	private MessageDispatcher messageDispatcher;
	
	private View view;
	
	private final boolean neverBeMaster;
	
	private Boolean rebuilding = false;
	
	private Thread masterWatcher = new Thread(this);
	
	private String channelName;
	
	private final int sleepTime = 2000;
	
	private List<Address> cachedSlaveRoleList;
	
	private Address cachedMaster;
	
	private boolean changed = true;
	
	
	private Logger logger = Logger.getLogger(Communicator.class);
	
	public Communicator(String channelName) {
		this(channelName, false);
	}
	
	public Communicator(String channelName, boolean neverBeMaster) {
		this.neverBeMaster = neverBeMaster;
		this.channelName = channelName;
		
		this.handlers = new ArrayList<RequestHandler>();
		
		Logger logger = Logger.getLogger(this.getClass());
		
		CommunicationHelper.checkSystemProperty(logger);
		
		try {
			channel = new JChannel("tcp.xml");
			messageDispatcher = new MessageDispatcher(channel, null, this, this);
			channel.connect(channelName);
		} catch (ChannelException e) {
			logger.error("Error occurs while creating the communicator.", e);
		}
		
		if (this.neverBeMaster) {
			masterWatcher.start();
		}
	}


	
	public void close() {
		if (this.neverBeMaster) {
			masterWatcher.stop();
		}
		channel.close();
		messageDispatcher.stop();
	}

	@Override
	public Object handle(Message msg) {
		ParameterizedMessage ParameterizedMessage = 
			(ParameterizedMessage) msg.getObject();
		
		for(RequestHandler handler : handlers) {
			if (handler.isCorresponding(ParameterizedMessage)) {
				Object ret = handler.handle(ParameterizedMessage);
//				if (!(ret instanceof ParameterizedMessage)) {
//					ret = buildErrorMsg(ret);
//				}
				return ret;
			}
		}
		
		// default msg
		ParameterizedMessageImpl defaultMsg = new ParameterizedMessageImpl("DefaultMSG");
		defaultMsg.setParameter("result", "ignored");
		
		return defaultMsg;
	}


	@Override
	public void block() {
	}

	@Override
	public void suspect(Address arg0) {
	}

	@Override
	public void viewAccepted(View newView) {
		this.view = newView;
		changed = true;
	}
	
	public Address getMaster() {
		return view.getCreator();
	}
	
	public List<Address> getSlaves() {
		Vector<Address> members = view.getMembers();
		return members.subList(1, members.size());
	}
	
	public List<Address> getAllMembers() {
		return view.getMembers();
	}
	
	public Address getLocalAddress() {
		return messageDispatcher.getChannel().getAddress();
	}
	

	
	public ParameterizedMessage sendMessage(Address dest, ParameterizedMessage msg) {
		checkConnectivity();
		
		Message message = new Message(dest, null, msg);
		
		Object response = null;
		try {
			response = this.messageDispatcher.sendMessage(message, RequestOptions.SYNC);
		} catch (TimeoutException e) {
			e.printStackTrace();
		} catch (SuspectedException e) {
			e.printStackTrace();
		}
		
		if (response instanceof ParameterizedMessage) {
			return (ParameterizedMessage) response;
		} else {
			return CommunicationHelper.buildErrorMsg(response);
		}
	}
	
	public ParameterizedMessage sendMessageToMaster(ParameterizedMessage msg) {
		return sendMessage(getMaster(), msg);
	}
	
	public GroupResponse castMessage(List<Address> dests, ParameterizedMessage msg) {
		checkConnectivity();
		
		Message messageToSend = new Message(null, null, msg);
		RspList rspList = this.messageDispatcher.castMessage(dests, messageToSend, RequestOptions.SYNC);
		
		return new GroupResponse(rspList);
	}
	
	public GroupResponse castMessage(ParameterizedMessage msg) {
		return castMessage(getAllMembers(), msg);
	}
	
	public void addRequestHandler(RequestHandler handler) {
		this.handlers.add(handler);
	}
	
	public boolean isMaster() {
		return this.getLocalAddress().equals(this.getMaster());
	}
	
	public void addRole(Role role) {
		List<RequestHandler> handlers = role.getHandlers();
		
		for(RequestHandler handler : handlers) {
			addRequestHandler(handler);
		}
	}
	
	public List<Address> getSlavesByRole() {
		if (changed) {
			updateCaches();
		}
		
		return cachedSlaveRoleList;
	}
	
	public synchronized void updateCaches() {
		if (changed) {
			updateSlavesCache();
			updateMasterCache();
		}
		
		changed = false;
	}
	
	public Address getMasterByRole() {
		if (changed) {
			updateCaches();
		}
		
		return cachedMaster;
	}

	private void updateMasterCache() {
		ParameterizedMessage msg = new ParameterizedMessageImpl("Role::MasterInfo");
		GroupResponse groupRsp = this.castMessage(msg);
		
		List<Address> masters = new ArrayList<Address>();

		for(Response rsp : groupRsp.getResponses()) {
			ParameterizedMessage responseMsg = rsp.getMessage();
			
			String isMaster = MessageHelper.getNonNullParameter(responseMsg, "isMaster");
			if (isMaster.equals("true")) {
				//cachedMaster = rsp.getSender();
				//return;
				masters.add(rsp.getSender());
			}
		}
		
		// Stable master order
		for(Address address : this.getAllMembers()) {
			if (masters.contains(address)) {
				cachedMaster = address;
				return;
			}
		}
	}

	private void updateSlavesCache() {
		ParameterizedMessage msg = new ParameterizedMessageImpl("Role::SlaveInfo");
		GroupResponse groupRsp = this.castMessage(msg);
		List<Address> ret = new ArrayList<Address>();
		for(Response rsp : groupRsp.getResponses()) {
			ParameterizedMessage responseMsg = rsp.getMessage();
			
			String isSlave = MessageHelper.getNonNullParameter(responseMsg, "isSlave");
			if (isSlave.equals("true")) {
				ret.add(rsp.getSender());
			}
		}
		cachedSlaveRoleList = ret;
	}
	
	private void checkConnectivity() {
		while (rebuilding) {
			try {
				logger.info("Communicator is not connected, sleep " + sleepTime + " to wait for connection...");
				Thread.sleep(this.sleepTime);
			} catch (InterruptedException e) {
				logger.error("Error occurs when try to sleep.", e);
			}
		}
	}

	@Override
	public void run() {
		while (true) {
			if (!this.channel.isConnected()) {
				logger.debug("channel is already disconnected.");
				logger.debug("reconnecting to " + channelName);
				try {
					this.channel.connect(channelName);
					this.rebuilding = false;
				} catch (ChannelException e) {
					logger.error("Error occurs when connecting to channel...", e);
				}
			}
			
			sleep();
			
			if (this.channel.isConnected() && this.isMaster()) {
				logger.debug("Detected to be master, disconnecting...");
				this.rebuilding = true;
				this.channel.disconnect();
				sleep();
			}
		}
	}

	private void sleep() {
		try {
			Thread.sleep(sleepTime);
		} catch (InterruptedException e) {
			logger.error("Error", e);
		}
	}
}
